package bma.common.langutil.concurrent;

import java.util.ArrayList;
import java.util.List;
import java.util.PriorityQueue;
import java.util.concurrent.Executor;

import bma.common.langutil.core.DateTimeUtil;

/**
 * 基于队列请求处理的定时执行对象
 * 
 * @author 关中
 * @since 1.0 (imported from BMA.ESP Project)
 * 
 */
public class ProcessTimeWheel extends ProcessQueue {

	final org.slf4j.Logger log = org.slf4j.LoggerFactory
			.getLogger(ProcessTimeWheel.class);

	protected HashedTimeWheel wheel;
	protected int initTaskSize = 10;
	protected Executor executor;

	protected PriorityQueue<HashedTimeWheelItem> tasks;
	protected List<HashedTimeWheelItem> items;
	protected long lastWheelTime;
	protected long nextWheelTime;

	public ProcessTimeWheel() {
		super();
		skipLogWait = true;
	}

	public ProcessTimeWheel(String name) {
		super(name);
		skipLogWait = true;
	}

	public Executor getExecutor() {
		return executor;
	}

	public void setExecutor(Executor executor) {
		this.executor = executor;
	}

	public HashedTimeWheel getWheel() {
		return wheel;
	}

	public void setWheel(HashedTimeWheel wheel) {
		this.wheel = wheel;
	}

	public int getInitTaskSize() {
		return initTaskSize;
	}

	public void setInitTaskSize(int taskInitSize) {
		this.initTaskSize = taskInitSize;
	}

	@Override
	public boolean initProcess() {
		if (this.wheel == null)
			throw new NullPointerException("wheel");
		if (!super.initProcess()) {
			return false;
		}
		this.nextWheelTime = System.currentTimeMillis();
		tasks = new PriorityQueue<HashedTimeWheelItem>(initTaskSize,
				new HashedTimeWheel.HashedTimeWheelItemComparator());
		items = new ArrayList<HashedTimeWheelItem>(1000);
		return true;
	}

	@Override
	public long processRun() {
		if (lastWheelTime == 0)
			lastWheelTime = System.currentTimeMillis();
		long ntime = wheel.processTickItems(lastWheelTime, items);
		this.nextWheelTime = ntime;
		if (items.isEmpty() && System.currentTimeMillis() >= ntime) {
			lastWheelTime = ntime;
		} else {
			tasks.addAll(items);
			items.clear();
		}

		HashedTimeWheelItem task = tasks.peek();
		while (task != null) {
			if (task.isCancel()) {
				tasks.poll();
			} else {
				if (task.getTime() <= System.currentTimeMillis()) {
					tasks.poll();
					// handle it
					try {
						ProcessTimerTask ptt = ((ProcessTimerTask) task);
						if (executor == null || ptt.isShortLived()) {
							ptt.run();
						} else {
							executor.execute(ptt);
						}
					} catch (Exception e) {
						// skip
						if (log.isDebugEnabled()) {
							log.debug("timetask(" + task + ") process fail", e);
						}
					}
				} else {
					// this.skipLogWait = false;
					return task.getTime();
				}
			}
			// do next
			task = tasks.peek();
		}
		// this.skipLogWait = true;
		// log.debug("next time {}:{}", ntime, ntime -
		// System.currentTimeMillis());
		return ntime;
	}

	@Override
	public boolean handleIntent(Object intent) {
		if (intent instanceof ProcessTimerTask) {
			ProcessTimerTask task = (ProcessTimerTask) intent;
			if (task.isCancel()) {
				if (log.isDebugEnabled()) {
					log.debug("{}: remove task({})", getName(), task);
				}
				boolean r = this.tasks.remove(task);
				if (!r) {
					r = wheel.remove(task);
				}
				if (log.isDebugEnabled()) {
					log.debug("{}: remove => {}", getName(), r);
				}
			} else {
				if (task.getTime() <= this.nextWheelTime) {
					if (log.isDebugEnabled()) {
						log.debug("{}: direct task({})", getName(), task);
					}
					this.tasks.add(task);
				} else {
					if (log.isDebugEnabled()
							&& !(task instanceof ProcessTimerLogSilence)) {
						log.debug(
								"{}: wheel task({},{})",
								new Object[] {
										getName(),
										task,
										DateTimeUtil.formatPeriod(task
												.getTime()
												- System.currentTimeMillis()) });
					}
					wheel.add(task);
				}
			}
			return true;
		}
		return super.handleIntent(intent);
	}
}
